package com.cn.broker.processor;

import com.cn.broker.BrokerController;
import com.cn.common.processor.AbstractProcessor;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.common.topic.TopicData;
import com.cn.file.AppendMessageResult;
import com.cn.file.AppendMessageStatus;
import com.cn.file.MessageStore;
import com.cn.msg.Message;
import com.cn.util.RemotingSerializable;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.ObjectUtils;

/**
 * 消息存储
 */
public class MessageCommitProcessor extends AbstractProcessor implements INettyRequestProcessor {

    private BrokerController brokerController;

    public MessageCommitProcessor(BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    @Override
    public Message process(Message message, ChannelHandlerContext ctx) {
        MessageStore messageStore = RemotingSerializable.decode(message.getByteBodys(), MessageStore.class);
        TopicData topicData = this.brokerController.getTopicManager().getTopic(messageStore.getTopic());
        AppendMessageResult result = null;
        if (topicData == null) {
            result = new AppendMessageResult(AppendMessageStatus.TOPIC_NOT_FOUND);
        } else {
            messageStore.setSocketAddress(ctx.channel().remoteAddress());
            result = brokerController.getDefaultMessageStore().putMessage(messageStore);
        }
        if (result != null) {
            message.setByteBodys(RemotingSerializable.encode(result));//返回的消息
            message.setLength(message.getByteBodys().length);//长度
//            System.out.println("存储消息成功：" + result.getMsgId() + "==" + result.getStatus() + "==>opaque:" + message.getOpaque());
        }
        return message;
    }
}
